package spark.core.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object rddTransOperators {

  def main(args: Array[String]): Unit = {
    //TODO first. 建立和Spark框架的连接
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("spark-rddTransOperators")
    val sparkContext = new SparkContext(sparkConf)

    // Todo second. 业务过程代码
    // Todo 一、Value类型算子

    //    val dataRDD: RDD[Int] = sparkContext.makeRDD {
    //      List(1, 2, 3, 4)
    //    }

    // Todo 1. map
    // 将处理的数据逐条进行映射转换，这里的转换可以是类型的转换，也可以是值的转换。

    //    val dataRDD1: RDD[Int] = dataRDD.map(num => {
    //      num * 2
    //    })
    //    dataRDD1.collect().foreach(println)

    // Todo 2.mapPartitions
    // 将待处理的数据 以分区为单位 发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据。

    //    val dataRDD1: RDD[Int] = dataRDD.mapPartitions(
    //      datas => {
    //        datas.filter(_ == 2)
    //      }
    //    )
    //    dataRDD1.collect().foreach(println)

    // Map 算子是分区内一个数据一个数据的执行，类似于串行操作.
    // 而mapPartitions算子 是以分区为单位进行批处理操作。

    // Todo 3. mapPartitionsWithIndex
    // 将待处理的数据以分区为单位发送到计算节点进行处理，这里的处理是指可以进行任意的处理，哪怕是过滤数据，在处理时同时可以获取当前分区索引。

    //    val dataRDD1 = dataRDD.mapPartitionsWithIndex(
    //      (index, datas) => {
    //        datas.map(index,_)
    //      }
    //    )

    // Todo 4. flatMap
    // 将处理的数据进行扁平化后再进行映射处理，所以算子也称之为扁平映射

    //    val dataRDD = sparkContext.makeRDD(List(List(1, 2), List(3, 4)), 1)
    //    val dataRDD1 = dataRDD.flatMap(
    //      list => list
    //    )
    //    println(dataRDD1.collect().mkString("Array(", ", ", ")"))
    //    println(dataRDD1.collect().mkString(","))

    // Todo 5. glom
    // 将同一个分区的数据直接转换为相同类型的内存数组进行处理，分区不变

    //    val dataRDD = sparkContext.makeRDD(List(1, 2, 3, 4), 2)
    //    val glomRDD: RDD[Array[Int]] = dataRDD.glom()
    //    val maxRDD = glomRDD.map(
    //      array => {
    //        array.max
    //      }
    //    )
    //    glomRDD.collect().foreach(string => println(string.mkString(",")))
    //    maxRDD.collect().foreach(println)
    //    println(maxRDD.collect().sum)

    // Todo 6. groupBy
    // 将数据根据指定的规则进行分组, 分区默认不变，但是数据会被打乱重新组合，我们将这样的操作称之为 shuffle 。极限情况下，数据可能被分在同一个分区中
    // 一个组的数据在一个分区中，但是并不是说一个分区中只有一个组

    //    val rdd = sparkContext.makeRDD(List("Hello", "Spark", "Scala", "Hadoop"), 2)
    //    // 分组和分区没有必然的关系
    //    val groupRDD = rdd.groupBy(data => {
    //      data.charAt(0)
    //    })
    //    groupRDD.collect().foreach(println)

    // Todo 7. filter
    // 将数据根据指定的规则进行筛选过滤，符合规则的数据保留，不符合规则的数据丢弃。
    // 当数据进行筛选过滤后，分区不变，但是分区内的数据可能不均衡，生产环境下，可能会出现数据倾斜。

    //    val dataRDD = sparkContext.makeRDD(List(
    //      1, 2, 3, 4
    //    ), 1)
    //    val dataRDD1 = dataRDD.filter(_ % 2 == 0)
    //    dataRDD1.collect().foreach(println)

    // Todo 8. sample
    // 根据指定的规则从数据集中抽取数据

    //    val dataRDD = sparkContext.makeRDD(List(
    //      1, 2, 3, 4
    //    ), 1)
    //    // 抽取数据不放回（伯努利算法）
    //    // 伯努利算法：又叫0、1分布。例如扔硬币，要么正面，要么反面。
    //    // 具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要
    //    // 第一个参数：抽取的数据是否放回，false：不放回
    //    // 第二个参数：抽取的几率，范围在[0,1]之间,0：全不取；1：全取；
    //    // 第三个参数：随机数种子
    //    val dataRDD1 = dataRDD.sample(withReplacement = false, 0.5)
    //    // 抽取数据放回（泊松算法）
    //    // 第一个参数：抽取的数据是否放回，true：放回；false：不放回
    //    // 第二个参数：重复数据的几率，范围大于等于0.表示每一个元素被期望抽取到的次数
    //    // 第三个参数：随机数种子
    //    val dataRDD2 = dataRDD.sample(withReplacement = true, 2)
    //
    //    println(dataRDD1.collect().mkString(","))
    //    println(dataRDD2.collect().mkString(","))

    // Todo 9. distinct
    // 将数据集中重复的数据去重

    //    val dataRDD = sparkContext.makeRDD(List(
    //      1, 2, 3, 4, 1, 2
    //    ), 1)
    //    val dataRDD1 = dataRDD.distinct()
    //    val dataRDD2 = dataRDD.distinct(2)
    //
    //    println(dataRDD1.collect().mkString(","))
    //    println(dataRDD2.collect().mkString(","))

    // Todo 10. coalesce
    // 根据数据量缩减分区，用于大数据集过滤后，提高小数据集的执行效率
    // 当spark 程序中，存在过多的小任务的时候，可以通过coalesce方法，收缩合并分区，减少分区的个数，减小任务调度成本

    //    val rdd = sparkContext.makeRDD(List(1, 2, 3, 4, 5, 6), 3)
    //    // coalesce方法默认情况下 不会将分区的数据打乱重新组合 这种情况下的缩减分区可能会导致数据不均衡，出现数据倾斜
    //    // 如果想要让数据均衡，可以进行shuffle处理
    //    //val newRDD: RDD[Int] = rdd.coalesce(2)
    //    val newRDD: RDD[Int] = rdd.coalesce(2, shuffle = true)
    //
    //    newRDD.saveAsTextFile("output")

    // Todo 11. repartition
    // 该操作内部其实执行的是coalesce操作，参数shuffle的默认值为true。无论是将分区数多的RDD转换为分区数少的RDD，还是将分区数少的RDD转换为分区数多的RDD，repartition
    // 操作都可以完成，因为无论如何都会经shuffle过程。

    //    val dataRDD = sparkContext.makeRDD(List(
    //      1, 2, 3, 4, 1, 2
    //    ), 2)
    //    val dataRDD1 = dataRDD.repartition(4)

    // Todo 12. sortBy
    // 该操作用于排序数据。在排序之前，可以将数据通过f函数进行处理，之后按照f函数处理的结果进行排序，默认为升序排列。
    // 排序后新产生的RDD的分区数与原RDD的分区数一致。中间存在shuffle的过程

    //    val dataRDD = sparkContext.makeRDD(List(
    //      1, 2, 3, 4, 1, 2
    //    ), 2)
    //    val dataRDD1 = dataRDD.sortBy(num => num, ascending = false, 4)
    //    dataRDD1.collect().foreach(println)

    // Todo 二、双Value类型算子

    // Todo 13. intersection
    // 对源RDD和参数RDD求 交集 后返回一个新的RDD
    // 两个数据源要求 分区数量 要保持一致 + 两个数据源要求 分区中数据数量 保持一致

    //    val dataRDD1 = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val dataRDD2 = sparkContext.makeRDD(List(3, 4, 5, 6))
    //    val ans = dataRDD1.intersection(dataRDD2)
    //    ans.collect().foreach(println)

    // Todo 14. union
    // 对源RDD和参数RDD求 并集 后返回一个新的RDD

    //    val dataRDD1 = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val dataRDD2 = sparkContext.makeRDD(List(3, 4, 5, 6, 7))
    //    val dataRDD: RDD[Int] = dataRDD1.union(dataRDD2)
    //    dataRDD.collect().foreach(println)

    // Todo 15. subtract
    // 以一个RDD元素为主，去除两个RDD中重复元素，将其他元素保留下来。求 差集

    //    val dataRDD1 = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val dataRDD2 = sparkContext.makeRDD(List(3, 4, 5, 6))
    //    val dataRDD = dataRDD1.subtract(dataRDD2)
    //    dataRDD.collect().foreach(println)

    // Todo 16. zip
    // 将两个RDD中的元素，以键值对的形式进行 合并 。其中，键值对中的Key为第1个RDD中的元素，Value为第2个RDD中的相同位置的元素。

    //    val dataRDD1 = sparkContext.makeRDD(List(1, 2, 3, 4))
    //    val dataRDD2 = sparkContext.makeRDD(List(3, 4, 5, 6))
    //    val dataRDD: RDD[(Int, Int)] = dataRDD1.zip(dataRDD2)
    //    dataRDD.collect().foreach(println)

    // Todo 三、Key - Value类型算子

    // Todo 17. partitionBy
    // 将数据按照指定Partitioner重新进行分区。Spark默认的分区器是HashPartitioner

    //    val rdd: RDD[(Int, String)] = sparkContext.makeRDD(Array((1, "aaa"), (2, "bbb"), (3, "ccc")), 3)
    //    val rdd2: RDD[(Int, String)] = rdd.partitionBy(new HashPartitioner(2))

    // Todo 18. reduceByKey
    // 可以将数据按照相同的Key对Value进行聚合 (按照key 分组 + 聚合)

    //    val dataRDD1 = sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3),("a",2)))
    //    val dataRDD2 = dataRDD1.reduceByKey(_ + _)
    //    dataRDD2.collect().foreach(println)
    //    val dataRDD3 = dataRDD1.reduceByKey(_ + _, 2)

    // Todo 19. groupByKey
    // 将数据源的数据根据key对value进行分组 （仅分组）

    //    val dataRDD1 = sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3),("b",2)))
    //    val dataRDD: RDD[(String, Iterable[Int])] = dataRDD1.groupByKey()
    ////    dataRDD.collect().foreach(println)
    //    // 同基本转换操作中的map，只不过mapValues是针对[K,V]中的V值进行map操作
    //    val sumRDD: RDD[(String, Int)] = dataRDD.mapValues(iter => iter.sum)
    //    sumRDD.collect().foreach(println)
    //    val dataRDD3 = dataRDD1.groupByKey(2)
    //    val dataRDD4 = dataRDD1.groupByKey(new HashPartitioner(2))

    // TODO reduceByKey和groupByKey的区别？
    // 从shuffle 的角度:
    // reduceByKey 和 groupByKey 都存在shuffle 的操作.
    // 但是reduceByKey 可以在shuffle 前对分区内相同key的数据进行预聚合（combine）功能，
    // 这样会减少落盘的 数据量，而groupByKey只是进行分组，不存在数据量减少的问题，reduceByKey性能比较 高。

    // 从功能的角度:
    // reduceByKey其实包含分组和聚合的功能。GroupByKey只能分组，不能聚合.
    // 所以在分组聚合的场合下，推荐使用reduceByKey，如果仅仅是分组而不需要聚合。那么还是只能使用groupByKey

    // Todo 20. aggregateByKey
    // 将数据根据不同的规则进行 分区内计算 和 分区间计算
    // TODO : 取出每个分区内相同key的最大值然后分区间相加
    // aggregateByKey 算子是函数柯里化，存在两个参数列表
    // 1. 第一个参数列表中的参数表示初始值
    // 2. 第二个参数列表中含有两个参数
    // // 2.1 第一个参数表示分区内的计算规则
    // // 2.2 第二个参数表示分区间的计算规则
    //    val rdd = sparkContext.makeRDD(List(
    //      ("a", 1), ("a", 2), ("c", 3),
    //      ("b", 4), ("c", 5), ("c", 6)
    //    ), 2)
    //    // 0:("a",1),("a",2),("c",3) => (a,10)(c,10)
    //    //                                            => (a, 10)(b, 10)(c, 20)
    //    // 1:("b",4),("c",5),("c",6) => (b,10)(c,10)
    //    val resultRDD: RDD[(String, Int)] = rdd.aggregateByKey(10)(
    //      (x, y) => math.max(x, y),
    //      (x, y) => x + y
    //    )
    //    resultRDD.collect().foreach(println)

    // Todo 21. foldByKey
    // 当分区内计算规则和分区间计算规则相同时，aggregateByKey就可以简化为foldByKey

    //    val dataRDD1 = sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    //    val dataRDD2 = dataRDD1.foldByKey(0)(_ + _)

    // Todo 22. combineByKey
    // 最通用的对key-value型rdd进行聚集操作的聚集函数。类似于aggregate(),combineByKey()允许用户返回值的类型与输入不一致。

    //    val list: List[(String, Int)] = List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))
    //    val input: RDD[(String, Int)] = sparkContext.makeRDD(list, 2)
    //    val combineRdd: RDD[(String, (Int, Int))] = input.combineByKey(
    //      v => (v, 1),
    //      (acc: (Int, Int), v) => (acc._1 + v, acc._2 + 1),
    //      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    //    )
    //    combineRdd.mapValues {
    //      case (num, cnt) =>
    //        num / cnt
    //    }.collect().foreach(println)

    // Todo 思考 reduceByKey、foldByKey、aggregateByKey、combineByKey 的区别？
    // 1. reduceByKey: 相同 key 的第一个数据不进行任何计算，分区内和分区间计算规则相同
    // 2. FoldByKey: 相同 key的第一个数据和初始值进行分区内计算，分区内和分区间计算规则相同
    // 3. AggregateByKey：相同 key 的第一个数据和初始值进行分区内计算，分区内和分区间计算规 则可以不相同
    // 4. CombineByKey:当计算时，发现数据结构不满足要求时，可以让第一个数据转换结构。分区 内和分区间计算规则不相同。

    // Todo 23. sortByKey
    // 在一个(K,V)的RDD上调用，K必须实现Ordered接口(特质)，返回一个按照key进行排序的

    //    val dataRDD1 = sparkContext.makeRDD(List(("a", 1), ("b", 2), ("c", 3)))
    //    val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(ascending = false)
    //    sortRDD1.collect().foreach(println)
    //    val sortRDD1: RDD[(String, Int)] = dataRDD1.sortByKey(ascending = false)

    // Todo 24. join \ leftOuterJoin \ rightOuterJoin
    // 与 SQL 中inner join left join right join相同  在类型为(K,V)和(K,W)的RDD上调用

    // Todo 26. cogroup  == connect + group
    // 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD

    val dataRDD1 = sparkContext.makeRDD(List(("a", 1), ("a", 2), ("c", 3)))
    val dataRDD2 = sparkContext.makeRDD(List(("a", 1), ("c", 2), ("c", 3)))
    val cgRDD: RDD[(String, (Iterable[Int], Iterable[Int]))] = dataRDD1.cogroup(dataRDD2)
    val sumRDD: RDD[(String, Int)] = cgRDD.map {
      case (key, (iter1, iter2)) =>
        val sum1 = iter1.sum
        val sum2 = iter2.sum
        (key, sum1 + sum2)
    }
    sumRDD.collect().foreach(println)


    //    cgRDD.collect().foreach(println)

    // TODO last.关闭连接
    sparkContext.stop()
  }


}
